/*
 * thread.c
 *
 *  Created on: Oct 22, 2009
 *      Author: thienlong
 */

#include "server.h"
#include "config.h"
#include <event2/event.h>
#include <event2/http.h>
#include "log.h"
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>
#include <string.h>
#include "filecache.h"
#include "cache.h"
#ifdef BATCH_ACCEPT_MECHANISM
#define BATCH_SIZE 1024
#else
#define BATCH_SIZE 1
#endif

static struct thread_state *threads;
static int thread_counter; /* counter thread */
static pthread_mutex_t init_lock;
static pthread_cond_t init_cond;
static int last_thread = -1;
lru_cache *file_cache = NULL; /*the cache of file which file has relative path with webroot's path*/

#ifdef HAVE_SERVER_STATUS
struct server_status status;
#endif

/* declare functions which setup multithread mode*/
static void setup_thread(struct thread_state *thread, ServerConf *conf, int thread_id);
static void worker_handler(int fd, short event_flags, void *arg);
static void *worker_listen(void *arg);
static void file_checker_thread_init();
void evhttp_get_request(struct evhttp *, evutil_socket_t, struct sockaddr *,
		ev_socklen_t);

/*--------------------------IMPLEMENTATION------------------------------*/


void file_checker_thread_init(ServerConf *conf) {
	map_fn fns = {str_hash_fn, str_eq_fn, NULL, NULL};
	file_cache = lru_cache_new(&fns, conf->cacheSize, 16, load_content_file, conf->httpWebRoot, 64, NULL);
	pthread_attr_t thread_attr;
	pthread_attr_init(&thread_attr);
	pthread_t thread_id;
	//FIXME

	/*	if (pthread_create(&thread_id, &thread_attr, file_checker_handler, conf)
	 != 0) {
	 fprintf_log(LOG_ERROR,
	 "[%s] In %s() %d: can't create new file checker thread",
	 __FILE__, __FUNCTION__, __LINE__);
	 exit(1);
	 }*/
}

/* thread functions */
void worker_thread_init(ServerConf *conf) {
	file_checker_thread_init(conf);
	threads = (struct thread_state*) calloc(conf->httpThreadNum,
			sizeof(struct thread_state));
	if (!threads) {
		fprintf_log(
				LOG_ERROR,
				"[%s] In %s() %d: calloc() failed! can't allocate thread descriptors",
				__FILE__, __FUNCTION__, __LINE__);
		exit(1);
	}
	pthread_mutex_init(&init_lock, NULL);
	pthread_cond_init(&init_cond, NULL);
	int i = 0;

	/* setup thread */
	for (i = 0; i < conf->httpThreadNum; ++i) {
		setup_thread(&threads[i], conf, i);
	}

	pthread_mutex_lock(&init_lock);
	while (thread_counter < conf->httpThreadNum) {
		pthread_cond_wait(&init_cond, &init_lock);
	}
	pthread_mutex_unlock(&init_lock);
}

static void setup_thread(struct thread_state *thread, ServerConf *conf, int thread_id) {
	thread->base = event_base_new();
	if (!thread->base) {
		fprintf_log(
				LOG_ERROR,
				"[%s] In %s() %d: thread->base = NULL! can't allocate new event base",
				__FILE__, __FUNCTION__, __LINE__);
		exit(1);
	}

	//setup evhttp and call back function for http event
	thread->http = evhttp_new(thread->base);
	thread->thread_id = thread_id;
	if (!thread->base) {
		fprintf_log(
				LOG_ERROR,
				"[%s] In %s() %d: thread->http = NULL! can't allocate new http event",
				__FILE__, __FUNCTION__, __LINE__);
		exit(1);
	}

#ifdef HAVE_SERVER_STATUS
	evhttp_set_cb(thread->http, STATUS_URI, server_status, thread);
#endif
	evhttp_set_gencb(thread->http, server_handler_default, thread);
	evhttp_set_timeout(thread->http, conf->httpTimeout);

	int fds[2];
	if (pipe(fds) < 0) {
		fprintf_log(LOG_ERROR, "[%s] In %s() %d: pipe() failed", __FILE__,
				__FUNCTION__, __LINE__);
		exit(1);
	}
	thread->notify_receive_fd = fds[0];
	thread->notify_send_fd = fds[1];
	socket_queue_init(&thread->connect_sockets);
	lpool_init(&thread->cache);
	thread->notify_event = event_new(thread->base, thread->notify_receive_fd,
			EV_READ | EV_PERSIST, worker_handler, thread);
	if (event_add(thread->notify_event, NULL) < 0) {
		fprintf_log(LOG_ERROR,
				"[%s] In %s() %d: Can't monitor the event notify pipe",
				__FILE__, __FUNCTION__, __LINE__);
		exit(1);
	}
	pthread_attr_t thread_attr;
	pthread_attr_init(&thread_attr);
	pthread_t id;
	if (pthread_create(&id, &thread_attr, worker_listen, thread)
			!= 0) {
		fprintf_log(LOG_ERROR,
				"[%s] In %s() %d: can't create new worker thread", __FILE__,
				__FUNCTION__, __LINE__);
		exit(1);
	}
}

static void *worker_listen(void *arg) {
	//guarantees that thread resources are deallocated upon return
	pthread_detach(pthread_self());
	struct thread_state *thread = arg;
	pthread_mutex_lock(&init_lock);
	++thread_counter;
	pthread_cond_signal(&init_cond);
	pthread_mutex_unlock(&init_lock);
	event_base_dispatch(thread->base);
	return NULL;
}

static void worker_handler(int fd, short event_flags, void *arg) {
	struct thread_state *thread = arg;
	static char buf[BATCH_SIZE];
	int num_read = 0;
	if ((num_read = read(thread->notify_receive_fd, buf, BATCH_SIZE)) < 0) {
		fprintf_log(LOG_ERROR,
				"[%s] In %s() %d: can't read from libevent pipe", __FILE__,
				__FUNCTION__, __LINE__);
	}
	int i = 0;
	for (; i < num_read; ++i) {
		struct socket_item *sck = socket_queue_peek(&thread->connect_sockets); /* get pending connection from thread's connection queue */
		if (sck != NULL) {
			evhttp_get_request(thread->http, sck->fd,
					(struct sockaddr *) &sck->ss, sck->socklen);
			lpool_put_element(&thread->cache, SOCKET_ITEM_TYPE,
					(struct next_data*) sck);
		}
	}
}

struct thread_state* thread_get_request_handler(int nthreads) {
	++last_thread;
	last_thread = last_thread % nthreads;
	return threads + last_thread;
}

void dispatch_connection(struct socket_item *sck, struct thread_state *thread) {
	socket_queue_push(&thread->connect_sockets, sck);
	if (write(thread->notify_send_fd, "", 1) != 1) {
		fprintf_log(LOG_ERROR, "[%s] in %s() %d: Writing thread notify pipe",
				__FILE__, __FUNCTION__, __LINE__);
	}
}

/* socket queue functions */
void socket_queue_init(struct socket_queue *queue) {
	queue->head = NULL;
	queue->tail = NULL;
	pthread_mutex_init(&queue->lock, NULL);
	pthread_cond_init(&queue->cond, NULL);
}

struct socket_item *socket_queue_peek(struct socket_queue *queue) {
	struct socket_item *item = NULL;
	pthread_mutex_lock(&queue->lock);
	item = queue->head;
	if (item != NULL) {
		queue->head = (struct socket_item *) item->next;
		item->next = NULL;
		if (queue->head == NULL) {
			queue->tail = NULL;
		}
	}
	pthread_mutex_unlock(&queue->lock);
	return item;
}

struct socket_item *socket_queue_pop(struct socket_queue *queue) {
	struct socket_item *item = NULL;
	pthread_mutex_lock(&queue->lock);

	while (queue->head == NULL) {
		pthread_cond_wait(&queue->cond, &queue->lock);
	}
	item = queue->head;
	queue->head = (struct socket_item *) item->next;
	item->next = NULL;
	if (queue->head == NULL) {
		queue->tail = NULL;
	}
	pthread_mutex_unlock(&queue->lock);
	return item;
}

void socket_queue_push(struct socket_queue *queue, struct socket_item *item) {
	item->next = NULL;
	pthread_mutex_lock(&queue->lock);
	if (queue->tail == NULL) {
		queue->head = item;
	} else {
		queue->tail->next = (struct next_data *) item;
	}
	queue->tail = item;
	pthread_cond_signal(&queue->cond);
	pthread_mutex_unlock(&queue->lock);
}

//FIXME insert unuse object to cache
void socket_queue_clear(linked_pool *cache, struct socket_queue *queue) {
	pthread_mutex_lock(&queue->lock);
	struct next_data *item = (struct next_data *) queue->head;
	if (cache && item) {
		pthread_mutex_lock(&cache->sck_cache.lock);
		cache->sck_cache.tail->next = item;
		pthread_mutex_unlock(&cache->sck_cache.lock);
		queue->head = queue->tail = NULL;
	} else if (!cache && item) {
		socket_queue_free_item(queue);
	}
	pthread_mutex_unlock(&queue->lock);
}

void socket_queue_free_item(struct socket_queue *queue) {
	pthread_mutex_lock(&queue->lock);
	struct next_data *item = (struct next_data *) queue->head;
	struct next_data *next = NULL;
	while (item != NULL) {
		next = item->next;
		free(item);
		item = next;
	}
	queue->head = queue->tail = NULL;
	pthread_mutex_unlock(&queue->lock);
}

void socket_queue_free(struct socket_queue *queue) {
	socket_queue_free_item(queue);
	free(queue);
}

int socket_queue_size(struct socket_queue *queue) {
	int counter = 0;
	pthread_mutex_lock(&queue->lock);
	struct next_data *item = (struct next_data *) queue->head;
	while (item != NULL) {
		++counter;
		item = item->next;
	}
	pthread_mutex_unlock(&queue->lock);
	return counter;
}

struct socket_item *socket_item_new() {
	struct socket_item *item = calloc(1, sizeof(struct socket_item));
	if (item == NULL) {
		fprintf_log(LOG_ERROR, "[%s] In %s() %d: can't allocate socket_item",
				__FILE__, __FUNCTION__, __LINE__);
		exit(1);
	}
	return item;
}

void socket_item_free(struct socket_item *item) {
	free(item);
}

int server_init(int port, int pendNum) {
	int listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	if (listener < 0) {
		fprintf_log(LOG_ERROR, "[%s] In %s() %d: socket() failed", __FILE__,
				__FUNCTION__, __LINE__);
		exit(1);
	}
	int opt = 1;
	setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
	setsockopt(listener, SOL_SOCKET, SO_KEEPALIVE, &opt, sizeof(opt));
	evutil_make_socket_nonblocking(listener);
	evutil_make_socket_closeonexec(listener);
	struct sockaddr_in sin;
	sin.sin_family = AF_INET;
	sin.sin_addr.s_addr = htonl(INADDR_ANY);
	sin.sin_port = htons(port);
	if (bind(listener, (struct sockaddr*) &sin, sizeof(sin)) < 0) {
		int errsv = errno;
		fprintf_log(LOG_ERROR, "[%s] In %s() %d: bind() failed due %s", __FILE__,
				__FUNCTION__, __LINE__, strerror(errsv));
		exit(1);
	}
	if (listen(listener, pendNum) < 0) {
		int errsv = errno;
		fprintf_log(LOG_ERROR, "[%s] In %s() %d: listen() failed due %s", __FILE__,
				__FUNCTION__, __LINE__, strerror(errsv));
		exit(1);
	}
#ifdef HAVE_SERVER_STATUS
	//initiate server status
	server_status_init(&status);
#endif
	return listener;
}

int update_event(int fd, struct event *notify_event, struct event_base *base,
		short event_flags, event_handler handler, void *arg) {
	if (event_del(notify_event) == -1) {
		return -1;
	}
	event_assign(notify_event, base, fd, event_flags, handler, arg);
	if (event_add(notify_event, NULL) == -1) {
		return -1;
	}
	return 0;
}

void atomic_number_init(struct atomic_number *number) {
	number->value = 0;
	pthread_mutex_init(&number->lock, NULL);
}

void atomic_number_increase(struct atomic_number *number) {
	pthread_mutex_lock(&number->lock);
	number->value++;
	pthread_mutex_unlock(&number->lock);
}

void atomic_number_decrease(struct atomic_number *number) {
	pthread_mutex_lock(&number->lock);
	number->value--;
	pthread_mutex_unlock(&number->lock);
}

void atomic_number_set_number(struct atomic_number *number, long val) {
	pthread_mutex_lock(&number->lock);
	number->value = val;
	pthread_mutex_unlock(&number->lock);
}

long atomic_number_get_number(struct atomic_number *number) {
	long val = 0;
	pthread_mutex_lock(&number->lock);
	val = number->value;
	pthread_mutex_unlock(&number->lock);
	return val;
}

void server_status_init(struct server_status *status) {
	atomic_number_init(&status->conn_num);
	atomic_number_init(&status->req_num);
	status->start_time = time(NULL);
}

void server_status_increase_conn(struct server_status *status) {
	atomic_number_increase(&status->conn_num);
}

void server_status_increase_req(struct server_status *status) {
	atomic_number_increase(&status->req_num);
}
long server_status_get_conn(struct server_status *status) {
	return atomic_number_get_number(&status->conn_num);
}
long server_status_get_req(struct server_status *status) {
	return atomic_number_get_number(&status->req_num);
}
time_t server_status_get_starttime(struct server_status *status) {
	return status->start_time;
}
